<?php
/*
 * @Descripttion: 系统提供的消息队列服务，如果不需要此服务，只需要把本文件改名（不以 'cwservice_' 开头）即可
 * @version:
 * @Author: ewgof
 * @Date: 2020-08-12 00:24:23
 * @LastEditors: ewgof
 * @LastEditTime: 2023-02-24 10:28:08
*/

use Workerman\Worker;
use cw\facade\Queue;

$redis = null;
global $redis;

// 创建一个Worker监听 5210 端口，使用 text 协议通讯, 用于抢占队列功能使用
$queue_worker = new Worker("text://0.0.0.0:5210");

// 启动4个进程对外提供服务
$queue_worker->count = 4;
$queue_worker->name = 'queue-consumer';

$queue_worker->onConnect = function ($connection) {
    echo '有连接进来了 。。。。' . PHP_EOL;
};

$queue_worker->onWorkerStart = function ($connection) {
    $queue_log_file_name = config('queue.redis.log_file_name') ?? 'queue';
    try {
        // 初始化 队列服务
        echo '消息消费者服务启动了 。。。 ' . PHP_EOL;
        Queue::connect();
        $queue_name = trim(config('queue.redis.name') ?? 'message:queue:list', ',') . ',message:queue:retry';
        $queue_name = trim($queue_name, ',');
        $lock_name = 'cw_sys_msg_queue_lock';


        while (true) {
            if (lock($lock_name)) {
                $msg = \cw\facade\Queue::pull($queue_name);
                unlock($lock_name);
                if (empty($msg)) {
                    $msg = '从redis队列服务中拉取消息失败：超时了。' . PHP_EOL;
                    echo $msg;
                    log_msg($msg, 'info', $queue_log_file_name);
                    continue;
                }
                execute($msg);
            } else {
                sleep(1);
                continue;
            }
        }
    } catch (\Exception $e) {
        $msg = '消息消费者服务执行出错, errmsg => ' . $e->getMessage() . ', file: ' . $e->getFile() . ', line: ' . $e->getLine();
        echo $msg . PHP_EOL;
        log_msg($msg, 'error', $queue_log_file_name);
    }
};

// 接收到队列抢占消息
$queue_worker->onMessage = function ($connection, $msg) {
    echo '接收到抢占式推送消息：' . $msg . PHP_EOL;
    execute(json_decode($msg, true));
};


$queue_worker->onWorkerStop = function ($connection) {
    global $redis;
    unset($redis);
};


$queue_worker->onClose = function ($connection) {
};


if (!function_exists('execute')) {
    function execute(array $data)
    {
        $queue_name = $data['key'] ?? '';  // 如果是从redis中拉取到的消息，queue_name 表示 key
        $queue_msg_arr = $data['value'] ?? $data;
        $queue_config = config('queue');
        $queue_log_file_name = $queue_config['redis']['log_file_name'] ?? 'queue';
        $msg = '队列消费者拿到一条消息， msg => ' . json_encode($queue_msg_arr, JSON_UNESCAPED_UNICODE) . PHP_EOL;
        echo $msg;
        log_msg($msg, 'info');

        $handler = $queue_msg_arr['handler'];

        if (empty($handler[0]) || empty($handler[1])) {
            $msg = '类名或者方法名为空' . PHP_EOL;
            echo $msg;
            log_msg($msg, 'error', $queue_log_file_name);
            return false;
        }

        // 不知道为什么，这里必须先取出来而不能直接使用
        $class = $handler[0];
        $method = $handler[1];

        if (!class_exists($class)) {
            $msg = '类 ' . $class . ' 不存在' . PHP_EOL;
            echo $msg;
            log_msg($msg, 'error', $queue_log_file_name);
            return false;
        }
        if (!method_exists($class, $method)) {
            $msg = '方法 ' . $method . ' 不存在' . PHP_EOL;
            echo $msg;
            log_msg($msg, 'error', $queue_log_file_name);
            return false;
        }

        $msg = '处理任务 ... ' . PHP_EOL;
        echo $msg;
        log_msg($msg, 'info');

        $rm = new ReflectionMethod($class, $method);
        if ($rm->isStatic()) {
            // 静态调用
            $rst = $class::$method($queue_msg_arr['params']);
        } else {
            // 非静态调用
            if (!empty($queue_msg_arr['init_params'])) {
                $rst = (new $class($queue_msg_arr['init_params']))->$method($queue_msg_arr['params']);
            } else {
                $rst = (new $class())->$method($queue_msg_arr['params']);
            }
        }

        if (false === $rst) {
            // 如果处理逻辑里面明确返回 false 说明任务执行失败，任务重新入队列
            if (0 == $queue_msg_arr['retry_times']) {
                // 无限重试
                push($queue_msg_arr['handler'], $queue_msg_arr['params'], ++$queue_msg_arr['retried_times'], 'cw_sys:message:queue:retry', $queue_msg_arr['init_params']);
            } elseif ($queue_msg_arr['retried_times'] < $queue_msg_arr['retry_times'] && $queue_msg_arr['retry_times'] > 0) {
                // 重试一定次数
                push($queue_msg_arr['handler'], $queue_msg_arr['params'], ++$queue_msg_arr['retried_times'], 'cw_sys:message:queue:retry', $queue_msg_arr['init_params']);
            }
        }

        $msg = '处理完成一个任务，结果 => ' . var_export($rst, true)  . ', queue_msg_arr => ' . var_export($queue_msg_arr, true) . PHP_EOL;
        echo $msg;
        log_msg($msg, 'info');
    }
}

// 如果不是在根目录启动，则运行runAll方法
if (!defined('GLOBAL_START')) {
    Worker::runAll();
}
